package com.jinzheng.qianhai.commont;

import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.serializer.SerializerFeature;
import com.jinzheng.qianhai.entity.UserSocket;
import com.jinzheng.qianhai.service.UserSocketService;
import com.jinzheng.qianhai.util.HexUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.group.ChannelGroup;
import io.netty.channel.group.DefaultChannelGroup;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.util.ReferenceCountUtil;
import io.netty.util.concurrent.GlobalEventExecutor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;

import java.time.LocalDateTime;

/**
 * netty服务端出参
 */
@Slf4j
public class DiscardServerHandlers extends ChannelInboundHandlerAdapter {

    //用于保存客户端的通道组
    private static final ChannelGroup clients = new DefaultChannelGroup(GlobalEventExecutor.INSTANCE);

    /**
     * 读取客户端发送来的消息
     */
    @Override
    public void channelRead(ChannelHandlerContext ctx, Object msg) {
        try {
            byte[] byteValue = (byte[]) msg;
            log.info("客户端发来的数据，地址：" + ctx.channel().remoteAddress() + " 内容：" + HexUtils.toHexString(byteValue));
            //这里调用service服务
            boolean checkCode = checkCode(byteValue);
            System.out.println("数据校验结果："+checkCode);
            if(checkCode){
                //给其他人转发消息
                for (Channel client : clients) {
                    if(client.equals(ctx.channel())){
                        client.writeAndFlush(byteValue);
                    }
                }
                new SendData(byteValue).invoke();
            }
        }  finally {
            ReferenceCountUtil.release(msg);
        }
    }

    /**
     * 接入客户端
     */
    @Override
    public void handlerAdded(ChannelHandlerContext ctx) throws Exception {
        log.info("新客户端连接：{}", ctx.channel().id().asShortText());
        clients.add(ctx.channel());
    }

    /**
     * 断开客户端
     */
    @Override
    public void handlerRemoved(ChannelHandlerContext ctx) throws Exception {
        log.info("客户端连接断开：{}", ctx.channel().id().asShortText());
        clients.remove(ctx.channel());
    }

    /**
     * 表示channel 处于活动状态
     */
    @Override
    public void channelActive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " 处于活动状态");
    }

    /**
     * 表示channel 处于不活动状态
     */
    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {
        System.out.println(ctx.channel().remoteAddress() + " 处于不活动状态");
    }

    /**
     * 监听客户端传入数据
     * @param ctx
     * @param evt
     * @throws Exception
     */
    @Override
    public void userEventTriggered(ChannelHandlerContext ctx, Object evt) throws Exception {
        if (evt instanceof IdleStateEvent) {
            IdleStateEvent event = (IdleStateEvent) evt;
            if (event.state() == IdleState.READER_IDLE) {
                log.info("连接超时关闭");
                ctx.channel().writeAndFlush("连接超时关闭，已经120秒未收到客户端的消息了！");
                ctx.channel().close();
            }
        } else {
            super.userEventTriggered(ctx, evt);
        }
    }

    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        // 出现异常就关闭
        cause.printStackTrace();
        //ctx.close();
        ctx.channel().close();
        clients.remove(ctx.channel());
    }

    /**
     * 数据校验
     * @param bytes
     * @return
     */
    public static boolean checkCode(byte[] bytes){
        if(null==bytes || bytes.length < 25 || bytes[0]!=0x23 || bytes[1]!=0x23){
            log.warn("数据格式错误：{}", org.apache.tomcat.util.buf.HexUtils.toHexString(bytes));
            return false;
        }
        byte checkCode = 0x00;
        for (byte bt:bytes){
            checkCode ^= bt;
        }
        if(checkCode!=0x00){
            log.warn("数据校验错误：{}", org.apache.tomcat.util.buf.HexUtils.toHexString(bytes));
            return false;
        }
        return true;
    }

    /**
     * 发送数据
     */
    private class SendData {
        private byte[] object1;

        public SendData(byte... object1) {
            this.object1 = object1;
        }

        public void invoke() {
            KafkaTopicNameProp kafkaTopicNameProp = SpringUtil.getBean(KafkaTopicNameProp.class);
            String topicName = StringUtils.EMPTY;
            if(null!=kafkaTopicNameProp && StringUtils.isNotBlank(kafkaTopicNameProp.getTopicName())){
                topicName = kafkaTopicNameProp.getTopicName();
            }else{
                topicName = "JZQH_SOCKET_RECEIVE_VALUE";
            }
            String openstoreType = StringUtils.EMPTY;
            if(null!=kafkaTopicNameProp && StringUtils.isNotBlank(kafkaTopicNameProp.getOpenstoreType())){
                openstoreType = kafkaTopicNameProp.getOpenstoreType();
            }else{
                openstoreType = "storedb";
            }
            System.out.println(String.format("发送kafka监听topicName：%s，存储方式：%s",topicName,openstoreType));
            if("storekafka".equals(openstoreType)){
                KafkaProducer kafkaProducer = SpringUtil.getBean(KafkaProducer.class);
                String toJSONString = JSONObject.toJSONString(HexUtils.toHexString(object1), SerializerFeature.WriteMapNullValue);
                //kafkaProducer.sendMessageByTopic(topicName,null, JSONObject.toJSONString(object1, SerializerFeature.WriteMapNullValue));
                kafkaProducer.sendMessageByTopic(topicName,null, toJSONString);
            }else{
                String toJSONString = JSONObject.toJSONString(HexUtils.toHexString(object1), SerializerFeature.WriteMapNullValue);
                UserSocketService userSocketService = SpringUtil.getBean(UserSocketService.class);
                UserSocket userSocket = new UserSocket();
                userSocket.setTopicName(topicName);
                userSocket.setHeaderName("test_socket_header");
                userSocket.setNote(toJSONString);
                userSocket.setStatus(0);
                userSocket.setCreateTime(LocalDateTime.now());
                userSocketService.save(userSocket);
            }
        }
    }
}
